package cn.felord.debezium.debezium;
import cn.felord.debezium.dao.DetectionRlMapper;
import cn.felord.debezium.service.PushServiceNew;
import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Map;

import static io.debezium.data.Envelope.FieldName.*;
import static java.util.stream.Collectors.toMap;
/**
 * @since 2021 /6/1 17:01
 */
@Configuration
@Slf4j
public class DebeziumConfiguration {
    @Value("${debezuimdb.hostname}")
    String hostname;
    @Value("${debezuimdb.port}")
    Integer port;
    @Value("${debezuimdb.user}")
    String user;
    @Value("${debezuimdb.password}")
    String password;
    @Autowired
    DetectionRlMapper detectionRlMapper;
    @Autowired
    PushServiceNew pushServiceNew;
    /**
     * Debezium 配置.
     */
    @Bean
    io.debezium.config.Configuration debeziumConfig() {
        return io.debezium.config.Configuration.create()
//            连接器的Java类名称
                .with("connector.class", MySqlConnector.class.getName())
//            偏移量持久化，用来容错 默认值
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
//                偏移量持久化文件路径 默认/tmp/offsets.dat  如果路径配置不正确可能导致无法存储偏移量 可能会导致重复消费变更
//                如果连接器重新启动，它将使用最后记录的偏移量来知道它应该恢复读取源信息中的哪个位置。
                .with("offset.storage.file.filename", "/tmp/offsets.dat")
//                捕获偏移量的周期
                .with("offset.flush.interval.ms", "1")
//               连接器的唯一名称
                .with("name", "mysql-connector")
//                数据库的hostname
                .with("database.hostname", hostname)
//                端口
                .with("database.port", port)
//                用户名
                .with("database.user", user)
//                密码
                .with("database.password", password)
//                 包含的数据库列表
                .with("database.include.list", "hussar_mf")
                .with("table.include.list",
                        "hussar_mf.tb_detection_rl,"//熔炼质检
                        +"hussar_mf.tb_melting_measurement,"//测温记录
                        +"hussar_mf.tb_pouring_measurement,"//测温记录
                                + "hussar_mf.tb_upload_truck_feeding")//加料监控
//                是否包含数据库表结构层面的变更，建议使用默认值true
                .with("include.schema.changes", "false")
//                mysql.cnf 配置的 server-id
                .with("database.server.id", 1)
//                	MySQL 服务器或集群的逻辑名称
                .with("database.server.name", "customer-mysql-db-server")
//                历史变更记录
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
//                历史变更记录存储位置，存储DDL
                .with("database.history.file.filename", "/tmp/dbhistory.dat")
                .build();
    }
    /**
     * Debezium server bootstrap debezium server bootstrap.
     */
    @Bean
    DebeziumServerBootstrap debeziumServerBootstrap(io.debezium.config.Configuration configuration) {
        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
        DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(configuration.asProperties())
                .notifying(this::handlePayload)
                .build();
        debeziumServerBootstrap.setDebeziumEngine(debeziumEngine);
        return debeziumServerBootstrap;
    }
    private void handlePayload(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) {
        recordChangeEvents.forEach(r -> {
            SourceRecord sourceRecord = r.record();
            String topic = sourceRecord.topic();
            Struct sourceRecordChangeValue = (Struct) sourceRecord.value();
            if (sourceRecordChangeValue != null) {
                // 判断操作的类型 过滤掉读 只处理增删改   这个其实可以在配置中设置
                Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
                if (operation != Envelope.Operation.READ) {
                    String record = operation == Envelope.Operation.DELETE ? BEFORE : AFTER;
                    // 获取增删改对应的结构体数据
                    Struct struct = (Struct) sourceRecordChangeValue.get(record);
                    // 将变更的行封装为Map
                    Map<String, Object> payload = struct.schema().fields().stream()
                            .map(Field::name)
                            .filter(fieldName -> struct.get(fieldName) != null)
                            .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                            .collect(toMap(Pair::getKey, Pair::getValue));
                    // 这里简单打印一下
                    System.out.println("operation = " + operation);
//                    System.out.println("data = " + payload);
                    if(operation.toString().equals("CREATE")){
                        String id =  payload.get("id").toString();
                        System.out.println("新增记录一条,id:" + payload.get("id"));
                        if(topic.split("\\.").length > 2){
                            String tableName = topic.split("\\.")[2];
                            System.out.println("tabelName : " + tableName);
                            pushServiceNew.receive(tableName,id);
                        }else {
                            log.error("debezuim解析表i名字异常");
                        }

                    }else if(operation.toString().equals("UPDATE")){

                    }
                }
            }
        });
    }

}
